package cn.icanci.loopstack.rec.engine.sdk.server;

import cn.hutool.http.Method;
import cn.hutool.json.JSONUtil;
import cn.icanci.loopstack.rec.engine.sdk.properties.RecProperties;
import cn.icanci.loopstack.rec.engine.sdk.rule.repository.EngineRepositoryHolder;
import cn.icanci.loopstack.rec.engine.sdk.spi.DomainSPI;
import cn.icanci.loopstack.rec.common.model.socket.RegisterDTO;
import cn.icanci.loopstack.rec.common.model.socket.UriConstant;
import cn.icanci.loopstack.rec.common.result.R;
import cn.icanci.loopstack.rec.common.utils.IPUtils;
import cn.icanci.loopstack.rec.engine.script.client.Client;
import cn.icanci.loopstack.rec.engine.script.client.RemoteException;
import cn.icanci.loopstack.rec.engine.script.client.http.HttpClientImpl;

import java.util.Set;
import java.util.concurrent.*;

import javax.annotation.Resource;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

/**
 * @author icanci
 * @since 1.0 Created in 2022/11/20 21:29
 */
@Service
public class RegisterServer implements InitializingBean {
    private static final Logger             logger         = LoggerFactory.getLogger(RegisterServer.class);
    /** http实例 */
    private static final Client                 CLIENT         = HttpClientImpl.getInstance();
    @Resource
    private              RecProperties          recProperties;
    @Resource
    private              DomainSPI              domainSPI;
    @Resource
    private              EngineRepositoryHolder engineRepositoryHolder;

    private static final int                CORE_SIZE      = Runtime.getRuntime().availableProcessors();

    private static final ThreadPoolExecutor REGISTER_POOL  = new ThreadPoolExecutor(CORE_SIZE,             //
        CORE_SIZE << 1,                                                                                    //
        60L,                                                                                               //
        TimeUnit.SECONDS,                                                                                  //
        new LinkedBlockingQueue<>(2000),                                                                   //
        runnable -> new Thread(runnable, "RegisterClient Biz Pool-" + runnable.hashCode()),                //
        (r, executor) -> {
            throw new RuntimeException("RegisterClient Biz Pool is EXHAUSTED!");
        });

    /** 注册请求地址 */
    private static final String             REQ_URL_FORMAT = "http://%s:%s%s";

    /**
     * 将SDK所在服务注册到注册中心
     *
     * @param serverAddress 服务端ip地址
     * @param serverPort 服务端端口
     * @param appName 客户端服务名
     */
    public void register(String serverAddress, int serverPort, int clientPort, String appName) {
        String domain = recProperties.getDomain();
        if (recProperties.isLoadAll()) {
            Set<String> domainCodes = domainSPI.loadAllDomainCodes();
            domain = String.join(",", domainCodes);
        }
        String[] addresses = serverAddress.split(",");

        // TODO 注册地址打散，防止压力放在同一个机器上
        for (String address : addresses) {
            try {
                FutureTask<R> task = new FutureTask<>(new RegisterCallable(address, clientPort, appName, domain, serverPort));
                REGISTER_POOL.execute(task);
                R r = task.get(10, TimeUnit.SECONDS);
                if (r.isOk()) {
                    break;
                } else {
                    logger.error("Task Register Exception:{}", r.getMessage());
                }
            } catch (RemoteException | ExecutionException | InterruptedException | TimeoutException e) {
                logger.warn("Register Exception:{}", e.getMessage());
            }
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (!recProperties.isLoad()) {
            return;
        }
        // 项目启动加载缓存
        refreshCache();

        // 注入注册bean
        RecNettyServer.setRegisterService(this);

        // 启动服务器
        RecNettyServer.startClient(recProperties.getServerIps(), recProperties.getServerPort(), recProperties.getClientPort(), recProperties.getAppName());
    }

    private void refreshCache() {
        boolean loadAll = recProperties.isLoadAll();
        if (loadAll) {
            Set<String> domainCodes = domainSPI.loadAllDomainCodes();
            engineRepositoryHolder.refresh(domainCodes);
            return;
        }

        String domain = recProperties.getDomain();
        if (StringUtils.isNotBlank(domain)) {
            Set<String> domainCodes = Sets.newHashSet(domain.replaceAll("\\s*", StringUtils.EMPTY).split(","));
            engineRepositoryHolder.refresh(domainCodes);
        }
    }

    /** 注册器  */
    private static class RegisterCallable implements Callable<R> {

        private final String address;
        private final int    clientPort;
        private final String appName;
        private final String domain;
        private final int    serverPort;

        public RegisterCallable(String address, int clientPort, String appName, String domain, int serverPort) {
            this.address = address;
            this.clientPort = clientPort;
            this.appName = appName;
            this.domain = domain;
            this.serverPort = serverPort;
        }

        @Override
        public R call() throws Exception {
            RegisterDTO registerDTO = new RegisterDTO(IPUtils.getHostIpAddress(), clientPort, appName, domain);
            String reqUrl = String.format(REQ_URL_FORMAT, address, serverPort, UriConstant.REGISTER);

            Client.RpcRequest rpcRequest = new Client.RpcRequest(reqUrl, registerDTO, Maps.newHashMap(), Method.POST, 3, TimeUnit.SECONDS, 3);

            R call = CLIENT.call(rpcRequest, R.class);

            logger.info("[RegisterCallable][call] Register result:{}", JSONUtil.toJsonStr(call));

            return call;
        }
    }
}
